package com.hzgj.bcl.lock;

import com.google.common.base.Charsets;
import com.google.common.collect.Maps;
import com.hzgj.bcl.redis.LockEntry;
import com.hzgj.bcl.redis.RedisClient;
import com.hzgj.bcl.redis.RedisSubScribeListener;
import com.hzgj.bcl.util.lang.ThreadPoolKit;
import lombok.Getter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;

/**
 * Distributed implementation of {@link java.util.concurrent.locks.Lock}
 * Implements reentrant lock.<br>
 * Lock will be removed automatically if client disconnects.
 * <p>
 * Implements a <b>non-fair</b> locking so doesn't guarantees an acquire order.
 */
public class RedisLock implements RLock {


    private static final Logger logger = LoggerFactory.getLogger(RedisLock.class);

    public static final long LOCK_EXPIRATION_INTERVAL_SECONDS = 30;
    private static final ConcurrentMap<String, ScheduledFuture<?>> expirationRenewalMap = Maps.newConcurrentMap();
    protected final ConcurrentMap<String, LockEntry> entries = Maps.newConcurrentMap();
    protected long internalLockLeaseTime = TimeUnit.SECONDS.toMillis(LOCK_EXPIRATION_INTERVAL_SECONDS);
    public static final String unlockMessage = "0";

    final UUID id;
    protected final RedisClient redisClient;
    @Getter
    final String name;

    public RedisLock(RedisClient redisClient, String name, UUID id) {
        this.redisClient = redisClient;
        this.id = id;
        this.name = name;
    }

    @Override
    public void lock() {
        try {
            lockInterruptibly();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    @Override
    public void lock(long leaseTime, TimeUnit unit) {
        try {
            lockInterruptibly(leaseTime, unit);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }


    @Override
    public void lockInterruptibly() throws InterruptedException {
        lockInterruptibly(-1, null);
    }

    @Override
    public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
        long threadId = Thread.currentThread().getId();
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return;
        }
        LockEntry entry = subscribe(threadId);
        try {
            while (true) {
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    break;
                }

                // waiting for message
                if (ttl >= 0) {
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    getEntry(threadId).getLatch().acquire();
                }
            }
        } finally {
            unsubscribe(entry);
        }
    }

    @Override
    public boolean tryLock() {
        Long ttl = tryAcquire(-1, null, Thread.currentThread().getId());
        if (ttl == null) {
            return true;
        } else {
            return false;
        }
    }

    @Override
    public boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException {
        return tryLock(waitTime, -1, unit);
    }

    @Override
    public Condition newCondition() {
        // TODO implement
        throw new UnsupportedOperationException();
    }

    @Override
    public void unlock() {
        Boolean opStatus = unlockInner(Thread.currentThread().getId());
        if (opStatus == null) {
            throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                    + id + " thread-id: " + Thread.currentThread().getId());
        }
        if (opStatus) {
            cancelExpirationRenewal();
        }
    }

    @Override
    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        final long threadId = Thread.currentThread().getId();
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return true;
        }

        time -= (System.currentTimeMillis() - current);
        if (time <= 0) {
            return false;
        }

        current = System.currentTimeMillis();
        LockEntry entry = subscribe(threadId);
        try {
            time -= (System.currentTimeMillis() - current);
            if (time <= 0) {
                return false;
            }

            while (true) {
                long currentTime = System.currentTimeMillis();
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    return true;
                }

                time -= (System.currentTimeMillis() - currentTime);
                if (time <= 0) {
                    return false;
                }

                // waiting for message
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }

                time -= (System.currentTimeMillis() - currentTime);
                if (time <= 0) {
                    return false;
                }
            }
        } finally {
            unsubscribe(entry);
        }
    }

    private LockEntry subscribe(long threadId) {
        RedisSubScribeListener listener = new RedisSubScribeListener() {
            @Override
            public void onMessage(String channel, String message) {
                if (unlockMessage.equals(message)) {
                    getEntry(threadId).getLatch().release();
                }
            }
        };
        Thread thread = new Thread(() -> {
            redisClient.subscribe(listener, getChannelName());
            logger.info("subscribe out.");
        });
        thread.setDaemon(true);
        thread.start();
        LockEntry entry = LockEntry.builder().redisSubScribeListener(listener).build();
        entries.putIfAbsent(getEntryName(), entry);
        return entry;
    }

    private void unsubscribe(LockEntry entry) {
        boolean removed = entries.remove(getEntryName()) == entry;
        if (!removed) {
            throw new IllegalStateException();
        }
        entry.getRedisSubScribeListener().unsubscribe(getChannelName());
    }

    private Long tryAcquire(long leaseTime, TimeUnit unit, final long threadId) {
        if (leaseTime != -1) {
            return tryLockInner(leaseTime, unit, threadId);
        }
        Long ttlRemaining = tryLockInner(LOCK_EXPIRATION_INTERVAL_SECONDS, TimeUnit.SECONDS, threadId);
        if (ttlRemaining == null) {
            //获得锁
            scheduleExpirationRenewal(threadId);
        }
        return ttlRemaining;
    }

    private void scheduleExpirationRenewal(final long threadId) {
        if (expirationRenewalMap.containsKey(getEntryName())) {
            return;
        }

        ScheduledExecutorService scheduledExecutorService = ThreadPoolKit.getScheduledExecutorService("scheduleExpirationRenewal of " + getName());
        Runnable task = () -> {
            try {
                Long value = (Long) redisClient.eval(
                        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                                    "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                                    "return 1; " +
                              "end; " +
                              "return 0;",
                        Collections.singletonList(getName()),
                        Arrays.asList(Long.toString(internalLockLeaseTime), getLockName(threadId)));
                expirationRenewalMap.remove(getEntryName());
                if (Long.compare(value, 1L) == 0) {
                    scheduleExpirationRenewal(threadId);
                }
            } catch (Exception e) {
                logger.error("Can't update lock " + getName() + " expiration", e);
                return;
            }
        };

        ScheduledFuture<?> scheduledFuture = scheduledExecutorService.schedule(task, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
        if (expirationRenewalMap.putIfAbsent(getEntryName(), scheduledFuture) != null) {
            scheduledFuture.cancel(true);
        }
    }

    private void cancelExpirationRenewal() {
        ScheduledFuture<?> scheduledFuture = expirationRenewalMap.remove(getEntryName());
        if (scheduledFuture != null) {
            scheduledFuture.cancel(true);
        }
    }

    private Long tryLockInner(long leaseTime, TimeUnit unit, long threadId) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        try {
            return (Long) redisClient.eval(
                    "if (redis.call('exists', KEYS[1]) == 0) then " +
                                "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                                "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                                "return nil; " +
                            "end; " +
                            "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                                "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                                "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                                "return nil; " +
                            "end; " +
                            "return redis.call('pttl', KEYS[1]);",
                    Arrays.asList(getName()), Arrays.asList(Long.toString(internalLockLeaseTime), getLockName(threadId)));
        } catch (Exception e) {
            logger.error("tryLockInner {}", e);
            return -1L;
        }
    }

    private Boolean unlockInner(long threadId) {
        try {
            Long result = (Long) redisClient.eval(
                    "if (redis.call('exists', KEYS[1]) == 0) then " +
                                "redis.call('publish', KEYS[2], ARGV[1]); " +
                                "return 1; " +
                            "end;" +
                            "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                                "return nil;" +
                            "end; " +
                            "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                            "if (counter > 0) then " +
                                "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                                "return 0; " +
                            "else " +
                                "redis.call('del', KEYS[1]); " +
                                "redis.call('publish', KEYS[2], ARGV[1]); " +
                                "return 1; " +
                            "end; " +
                            "return nil;",
                    Arrays.asList(getName(), getChannelName()), Arrays.asList(unlockMessage, Long.toString(internalLockLeaseTime), getLockName(threadId)));
            return result != null ? (result == 0 ? false : true) : null;
        } catch (Exception e) {
            logger.error("unlockInner {}", e);
        }
        return false;
    }

    protected LockEntry getEntry(long threadId) {
        return entries.get(getEntryName());
    }


    @Override
    public boolean forceUnlock() {
        cancelExpirationRenewal();
        try {
            Long value = (Long) redisClient.eval(
                    "if (redis.call('del', KEYS[1]) == 1) then "
                                + "redis.call('publish', KEYS[2], ARGV[1]); "
                                + "return 1 "
                            + "else "
                                + "return 0 "
                            + "end",
                    Arrays.asList(getName(), getChannelName()), Arrays.asList(unlockMessage));
            if (value == 1) {
                return true;
            }
        } catch (Exception e) {
            logger.error("forceUnlock {}", e);
        }
        return false;
    }

    @Override
    public boolean isLocked() {
        return redisClient.exist(getName());
    }

    @Override
    public boolean isHeldByCurrentThread() {
        return redisClient.hexists(getName().getBytes(Charsets.UTF_8), getLockName(Thread.currentThread().getId()).getBytes(Charsets.UTF_8));
    }

    @Override
    public int getHoldCount() {
        byte[] result = redisClient.hget(getName().getBytes(Charsets.UTF_8), getLockName(Thread.currentThread().getId()).getBytes(Charsets.UTF_8));
        if (result == null) {
            return 0;
        }
        return Integer.parseInt(new String(result, Charsets.UTF_8));
    }

    protected String getEntryName() {
        return id + ":" + getName();
    }

    String getChannelName() {
        return prefixName("redis_lock__channel", getName());
    }

    protected String prefixName(String prefix, String name) {
        if (name.contains("{")) {
            return prefix + ":" + name;
        }
        return prefix + ":{" + name + "}";
    }

    String getLockName(long threadId) {
        return id + ":" + threadId;
    }
}
